package KeyedVSNo_KeyedWindows;

import bean.WaterSensor;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Collection;

/**
 * @author Spring_Hu
 * @date 2021/10/13 20:49
 */
public class KeyedVSNo_Keyed {
    public static void main(String[] args) throws Exception {
        //keyedby和keyby的区别 不keyedby 直接allwindow只能有一个并行实例即 一个并行度
        /*
        * <p>Note: This operation is inherently non-parallel since all elements have to pass through
            the same operator instance.
        *
        * */
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        //  .readTextfile 并行度也只能是1
        //  普通的SourceFunction的实现类是不能够修改并行度的 只有 ParallelSourceFunction 才可以修改并行度
        //使用webUI的stream
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.socketTextStream("localhost",8888)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],Long.parseLong(split[1]),Integer.parseInt(split[2]));
                    }
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        out.collect(Arrays.asList(elements).toString());
                    }
                }).print("print");

        env.execute();
    }
}
